package com.lboyang.remoting.transport.netty.codec;

import com.lboyang.common.enums.CompressTypeEnum;
import com.lboyang.common.enums.SerializationTypeEnum;
import com.lboyang.common.extension.ExtensionLoader;
import com.lboyang.remoting.compress.Compress;
import com.lboyang.common.constants.NrpcConstants;
import com.lboyang.common.dto.NrpcMessage;
import com.lboyang.common.dto.NrpcRequest;
import com.lboyang.common.dto.NrpcResponse;
import com.lboyang.remoting.serialize.Serializer;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;

import java.util.Arrays;

/**
 * @Description
 *      RPC 消息的解码器
 *      先看如何变码 RpcMessageEncoder ，明白了编码解码也就不难理解，就不每条都写注释了
 *      参考 <a href="https://zhuanlan.zhihu.com/p/95621344">
 * @Date 2021/9/12 0:00
 * @author: A.iguodala
 */
@Slf4j
public class RpcMessageDecoder extends LengthFieldBasedFrameDecoder {

    public RpcMessageDecoder() {
        /**
         * lengthFieldOffset:【魔数 4B】+【版本 1B】= 5B
         * lengthFieldLength: 长度字段是 4B
         * lengthAdjustment: 总长包括所有数据，之前读取 9 个字节，因此左侧长度为（fullLength-9）。所以值是 -9
         * initialBytesToStrip: 不需要跳过字符 0
         */
        this(NrpcConstants.MAX_FRAME_LENGTH, 5, 4, -9, 0);
    }

    /**
     * @param maxFrameLength      最大帧长度，决定可以接收的最大数据长度，超出则会被丢弃。
     * @param lengthFieldOffset   长度字段偏移量。长度字段是跳过指定字节长度的字段。
     * @param lengthFieldLength   长度字段的字节数。
     * @param lengthAdjustment    要添加到长度字段值的补偿值。
     * @param initialBytesToStrip 读取需要跳过的字节数。
     */
    public RpcMessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength,
                             int lengthAdjustment, int initialBytesToStrip) {
        super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip);
    }

    /**
     * 解码方法
     * @param ctx
     * @param in
     * @return
     * @throws Exception
     */
    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        Object decoded = super.decode(ctx, in);
        if (decoded instanceof ByteBuf) {
            ByteBuf frame = (ByteBuf) decoded;
            if (frame.readableBytes() >= NrpcConstants.TOTAL_LENGTH) {
                try {
                    return decodeFrame(frame);
                } catch (Exception e) {
                    log.error("Decode frame error!", e);
                    throw e;
                } finally {
                    frame.release();
                }
            }

        }
        return decoded;
    }


    /**
     * 缓冲区的内容解码为 RPC 消息
     * @param in
     * @return
     */
    private Object decodeFrame(ByteBuf in) {
        checkMagicNumber(in);
        checkVersion(in);
        int fullLength = in.readInt();
        byte messageType = in.readByte();
        byte codecType = in.readByte();
        byte compressType = in.readByte();
        int requestId = in.readInt();
        NrpcMessage rpcMessage = NrpcMessage.builder()
                .codec(codecType)
                .requestId(requestId)
                .messageType(messageType).build();
        if (messageType == NrpcConstants.HEARTBEAT_REQUEST_TYPE) {
            rpcMessage.setData(NrpcConstants.PING);
            return rpcMessage;
        }
        if (messageType == NrpcConstants.HEARTBEAT_RESPONSE_TYPE) {
            rpcMessage.setData(NrpcConstants.PONG);
            return rpcMessage;
        }
        int bodyLength = fullLength - NrpcConstants.HEAD_LENGTH;
        if (bodyLength > 0) {
            byte[] bs = new byte[bodyLength];
            in.readBytes(bs);
            String compressName = CompressTypeEnum.getName(compressType);
            Compress compress = ExtensionLoader.getExtensionLoader(Compress.class)
                    .getExtension(compressName);
            bs = compress.decompress(bs);
            String codecName = SerializationTypeEnum.getName(rpcMessage.getCodec());
            log.info("codec name: [{}] ", codecName);
            Serializer serializer = ExtensionLoader.getExtensionLoader(Serializer.class)
                    .getExtension(codecName);
            if (messageType == NrpcConstants.REQUEST_TYPE) {
                NrpcRequest tmpValue = serializer.deserialize(bs, NrpcRequest.class);
                rpcMessage.setData(tmpValue);
            } else {
                NrpcResponse tmpValue = serializer.deserialize(bs, NrpcResponse.class);
                rpcMessage.setData(tmpValue);
            }
        }
        return rpcMessage;

    }

    /**
     * 检查版本号
     * @param in
     */
    private void checkVersion(ByteBuf in) {
        byte version = in.readByte();
        if (version != NrpcConstants.VERSION) {
            throw new RuntimeException("version isn't compatible" + version);
        }
    }

    /**
     * 检查魔数
     * @param in
     */
    private void checkMagicNumber(ByteBuf in) {
        int len = NrpcConstants.MAGIC_NUMBER.length;
        byte[] tmp = new byte[len];
        in.readBytes(tmp);
        for (int i = 0; i < len; i++) {
            if (tmp[i] != NrpcConstants.MAGIC_NUMBER[i]) {
                throw new IllegalArgumentException("Unknown magic code: " + Arrays.toString(tmp));
            }
        }
    }
}
